package com.yanggu.flink.datastream_api.sink

import com.yanggu.flink.datastream_api.pojo.Book
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 * 自定义sink, 将数据输出到mysql中
 */
object JdbcSinkDemo02 {

  def main(args: Array[String]): Unit = {

    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    environment
      .fromCollection(
        Seq(Book(101L, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),
          Book(102L, "Streaming Systems2", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),
          Book(103L, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
          Book(105L, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017))
      )
      .addSink(new MyJdbcSink)

    environment.execute("JdbcSinkDemo02 Job")
  }

}

class MyJdbcSink extends RichSinkFunction[Book] {

  var conn: Connection = _

  var upsertStatement: PreparedStatement = _

  //在open声明周期方法中定义Connection和PreparedStatement
  //同时使用ON DUPLICATE KEY UPDATE的方式保持幂等, 这需要提供唯一索引或者是主键
  override def open(parameters: Configuration): Unit = {
    val tool = ParameterTool.fromPropertiesFile(getClass.getResourceAsStream("/jdbc.properties"))
    conn = DriverManager.getConnection(tool.get("jdbc.url"), tool.get("jdbc.username"), tool.get("jdbc.password"))
    upsertStatement = conn.prepareStatement("INSERT INTO books (id, title, authors, year) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE title = VALUES(title), authors = VALUES(authors), year = VALUES(year)")
  }

  override def invoke(value: Book, context: SinkFunction.Context): Unit = {
    upsertStatement.setLong(1, value.id)
    upsertStatement.setString(2, value.title)
    upsertStatement.setString(3, value.authors)
    upsertStatement.setInt(4, value.year)
    upsertStatement.execute()

    val count = upsertStatement.getUpdateCount
    println(s"受影响行数为: $count, 行")
    upsertStatement.clearParameters()
  }

  override def close(): Unit = {
    conn.close()
    upsertStatement.close()
  }

}